package org.example.aggreagte;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.example.data.WaterSensor;

public class ReduceDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<WaterSensor> source = env.fromElements(
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s1", 11L, 11),
                new WaterSensor("s1", 22L, 22),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s3", 3L, 3)
        );
        KeyedStream<WaterSensor, String> keyedStream = source.keyBy(WaterSensor::getId);
        SingleOutputStreamOperator<WaterSensor> reduce = keyedStream.reduce((v1, v2) -> {
            System.out.println("value1:" + v1);
            System.out.println("value2:" + v2);
            return new WaterSensor(v1.getId(), v1.getTs(), v1.getVc() + v2.getVc()
            );
        });
        reduce.print();

        env.execute("MapDemo");
    }
}
